package com.gd.data.platform.config.flink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Flink配置类
 * 配置Flink流处理环境
 * 
 * @author gd-data-platform
 * @since 1.0.0
 */
@Configuration
public class FlinkConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    /**
     * Flink流处理环境
     */
    @Bean
    public StreamExecutionEnvironment streamExecutionEnvironment() {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度
        env.setParallelism(2);

        // 启用Checkpoint
        env.enableCheckpointing(60000); // 1分钟检查点间隔

        return env;
    }

    /**
     * Kafka消费者配置
     */
    @Bean
    public KafkaSource<String> kafkaConsumer() {
        return KafkaSource.<String>builder()
                .setBootstrapServers(bootstrapServers)
                .setTopics("api-call-logs")
                .setGroupId(groupId)
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();
    }

    /**
     * Kafka生产者配置
     */
    @Bean
    public KafkaSink<String> kafkaProducer() {
        return KafkaSink.<String>builder()
                .setBootstrapServers(bootstrapServers)
                .setRecordSerializer(org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema.builder()
                        .setTopic("realtime-metrics")
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build())
                .build();
    }
}
